package com.seaboxdata.interfaceJob.queuing;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.seaboxdata.entity.queuing.VwQuDetailSource;
import com.seaboxdata.entity.queuing.VwQuDetailTarget;
import com.seaboxdata.kafka.BaseStringKafkaProducer;
import com.seaboxdata.service.queuing.VwQuDetailSourceService;
import com.seaboxdata.service.queuing.VwQuDetailTargetService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

@Component
@Configuration
@Slf4j
public class VwQuDetailJob {
    @Value("${queuing.VwQuDetail.Topic}")
    private String topic;
    @Autowired
    private BaseStringKafkaProducer baseStringKafkaProducer;

    @Autowired
    private VwQuDetailSourceService vwQuDetailSourceService;

    @Autowired
    private VwQuDetailTargetService vwQuDetailTargetService;

    @Value("${interface.isTest}")
    private String isTest;

    @Scheduled(cron = "${queuing.vwNetWork.Scheduled}")//每隔5分钟执行一次
    public void VwNetWorkreadKafka(){
        //如果是测试,直接退出
        if("true".equals(isTest)){
            return;
        }
        try {
            //查询目标表最后的更新时间
           // vwQuDetailTargetService.
            Date listUpdateTime = vwQuDetailTargetService.getListUpdateTime();
            LambdaQueryWrapper<VwQuDetailSource> queryWrapper1=new QueryWrapper<VwQuDetailSource>().lambda();
            if (listUpdateTime!=null){
                queryWrapper1.gt(VwQuDetailSource::getUpdateddate,listUpdateTime);
            }
            List<VwQuDetailSource> vwQuDetailSources = vwQuDetailSourceService.list(queryWrapper1);
            if (!CollectionUtils.isEmpty(vwQuDetailSources)) {
                //写入kafka
                Date date = new Date();
                for (VwQuDetailSource vwQuDetailSource : vwQuDetailSources) {
                    VwQuDetailTarget vwQuDetailTarget = new VwQuDetailTarget();
                    PropertyUtils.copyProperties(vwQuDetailTarget,vwQuDetailSource);
                    vwQuDetailTarget.setCjsj(date);
                    baseStringKafkaProducer.sendMessageToKafka(JSONObject.toJSONString(vwQuDetailTarget), topic);
                }
            }
        } catch (Exception e) {
            log.error("write vwNetWork error");
            e.printStackTrace();
        }
    }
}
